package com.hudsonmq.spring.impl.consume.management;

import java.util.concurrent.ConcurrentHashMap;

public class PushMessageMonitor {

    private static final long maxQueueMessage = 1000;

    /**
     * 更改每次拉取數據的時間點
     */
    private static final long changeConsumerSizeTime = 2000;


    private static final Integer maxPullMessageSize = 32;
    private static final Integer minPullMessageSize = 1;

    /**
     * 模擬邏輯阻塞隊列,根據 size 控制獲取數據的速率
     */
    private static ConcurrentHashMap<String, Integer> queueMessageMap = new ConcurrentHashMap<>();


    private static ConcurrentHashMap<String, Integer> pullMessageSizeMap = new ConcurrentHashMap<>();

    public static boolean isMax(String topic, int queueId) {
        Integer currentSize = queueMessageMap.computeIfAbsent(topic + queueId, k -> 0);
        return currentSize > maxQueueMessage;
    }

    public static void incrementQueueSize(String topic, int queueId, int size) {
        queueMessageMap.put(topic + queueId, queueMessageMap.computeIfAbsent(topic + queueId, k -> 0) + size);
    }

    public static void decrementQueueSize(String topic, int queueId, int size) {
        queueMessageMap.put(topic + queueId, Math.max(queueMessageMap.computeIfAbsent(topic + queueId, k -> 0) - size, 0));
    }

    public static void incrementPullMessageSize(String topic, int queueId) {
        Integer pullMessageSize = getPullMessageSize(topic, queueId);
        pullMessageSizeMap.put(topic + queueId, Math.min(pullMessageSize * 2, maxPullMessageSize));
    }

    public static void decrementPullMessageSize(String topic, int queueId) {
        Integer pullMessageSize = getPullMessageSize(topic, queueId);
        pullMessageSizeMap.put(topic + queueId, Math.max(pullMessageSize / 2, 1));
    }

    public static void changePullMessageSize(String topic, int queueId,long consumerTime) {
        if (consumerTime >= changeConsumerSizeTime){
            decrementPullMessageSize(topic,queueId);
        }else {
            incrementPullMessageSize(topic,queueId);
        }
        System.out.println(pullMessageSizeMap.get(topic+queueId));
    }

    public static Integer getPullMessageSize(String topic, int queueId) {
        return pullMessageSizeMap.computeIfAbsent(topic + queueId, k -> 32);
    }
}
